﻿using Confluent.Kafka;
using Grow.Binary;
using Grow.EventBus.Core.EventDataSender;
using Grow.EventBus.Core.Kafka.KafkaConnection;
using Grow.Module.Service;
using Grow.Utils.Exceptions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace Grow.EventBus.Core.Kafka
{
    internal class EventDataSend : IEventDataSend
    {
        private readonly IServiceContext _serviceContext;
        private readonly IKafkaConnection _kafkaConnection;
        private readonly IBinaryContext _binaryContext;
        public EventDataSend(IServiceContext serviceContext)
        {
            _serviceContext = serviceContext;
            _kafkaConnection= _serviceContext.GetRequiredService<IKafkaConnection>();
            _binaryContext = _serviceContext.GetRequiredService<IBinaryContext>();
        }

        public async  Task ExecuteAsync(EventDataDto eventDataDto, CancellationToken cancellationToken = default)
        {
            var producer = _kafkaConnection.GetProducer(eventDataDto.EventName);
            var value = _binaryContext.Serialize(eventDataDto);
            if (value is null)
            {
                throw new GrowCheckException($"[EventBus-Kafka] send msg fail, produce eventDataDto is null, {eventDataDto.EventDataId}");
            }
            var result = await producer.ProduceAsync(eventDataDto.EventName, new Message<string, byte[]>
            {
                Key = eventDataDto.EventDataId.ToString(),
                Value = value
            }, cancellationToken);

            if (!(result.Status is PersistenceStatus.Persisted or PersistenceStatus.PossiblyPersisted)) 
            {
                throw new GrowCheckException( $"[EventBus-Kafka] send msg fail, produce status  {result.Status}  {eventDataDto.EventDataId}");
            }
        }
    }
}
